package cn.iocoder.yudao.module.liveprice.service.message;

import cn.iocoder.yudao.module.liveprice.api.message.PriceMessage;
import cn.iocoder.yudao.module.liveprice.api.message.PriceMessageHandler;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 价格消息分发器
 */
@Slf4j
@Service
public class PriceMessageDispatcher {

    private final Map<String, PriceMessageHandler> handlers = new HashMap<>();
    private final ObjectMapper objectMapper;

    public PriceMessageDispatcher(List<PriceMessageHandler> handlerList, ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
        handlerList.forEach(handler -> handlers.put(handler.getMessageType(), handler));
    }

    /**
     * 分发处理消息
     *
     * @param messageContent Kafka接收的消息内容
     */
    public void dispatch(String messageContent) {
        try {
            // 解析基础消息以获取类型
            PriceMessage baseMessage = objectMapper.readValue(messageContent, PriceMessage.class);
            String messageType = baseMessage.getMessageType();
            
            PriceMessageHandler handler = handlers.get(messageType);
            if (handler == null) {
                log.warn("未找到消息类型[{}]的处理器", messageType);
                return;
            }
            
            // 根据具体类型解析为对应的消息对象
            Object specificMessage = objectMapper.readValue(messageContent, 
                    objectMapper.getTypeFactory().constructType(handler.getClass().getGenericInterfaces()[0]));
            
            // 处理消息
            handler.handleMessage(specificMessage);
        } catch (Exception e) {
            log.error("消息处理失败", e);
        }
    }
} 